// @Title business.go
// @Description cava smu服务主要业务逻辑
// @Author yezhibin 2021年5月1日01:22:46
// @Update

package business

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"time"

	"golang.org/x/text/encoding/simplifiedchinese"

	qfcfg "Projects/CavaCloudSys/cavateusrc/com.qfcfg"
	util "Projects/CavaCloudSys/cavateusrc/com.util"

	"github.com/go-basic/uuid"
	"github.com/pkg/errors"
	qflog "github.com/qf-tech/qftech-base/pkg/log"
	"nhooyr.io/websocket"
	"nhooyr.io/websocket/wsjson"
)

// Controller def
type Controller struct {
}

// Charset def
type Charset string

// const def
const (
	UTF8    = Charset("UTF-8")
	GB18030 = Charset("GB18030")
)

// Init def
func (control Controller) Init() {
	// 创建连接关闭通道
	closeCh = make(chan string, 1)

	// 初始化消息接收缓存通道，size=200，并启动接收消息协程
	recvMsgChan = make(chan RecvRemoteMsg, 200)
	go control.recvDataLoop(recvMsgChan, closeCh)

	// 初始化消息发送缓存通道，size=200，并启动发送消息协程
	sendMsgChan = make(chan SendRemoteMsg, 200)
	go control.sendDataLoop(sendMsgChan, closeCh)

	msgTypeMap = map[int]string{
		cavaClientRegisterReqMsg:    "cavaClientRegisterReqMsg",
		cavaClientRegisterRespMsg:   "cavaClientRegisterRespMsg",
		cavaTaskReqMsg:              "cavaTaskReqMsg",
		cavaTaskRespMsg:             "cavaTaskRespMsg",
		cavaTaskReportNotifyReqMsg:  "cavaTaskReportNotifyReqMsg",
		cavaTaskReportNotifyRespMsg: "cavaTaskReportNotifyRespMsg",
		cavaRegisterReqMsg:          "cavaRegisterReqMsg",
		cavaRegisterRespMsg:         "cavaRegisterRespMsg",
		cavaTaskStatusReqMsg:        "cavaTaskStatusReqMsg",
		cavaTaskStatusRespMsg:       "cavaTaskStatusRespMsg",
		cavaKeepAliveReqMsg:         "cavaKeepAliveReqMsg",
		cavaKeepAliveRespMsg:        "cavaKeepAliveRespMsg",
	}

	AliveCount = 0
}

func (control Controller) checkTraceID() string {
	uid := uuid.New()
	tmpSlice := strings.Split(uid, "-")
	slicelen := len(tmpSlice)
	traceID := tmpSlice[slicelen-1]
	return traceID
}

func (control Controller) genTraceID() string {
	uid := uuid.New()
	tmpSlice := strings.Split(uid, "-")
	slicelen := len(tmpSlice)
	traceID := tmpSlice[slicelen-1]
	return traceID
}

// transToStr def
func (control Controller) transToStr(v interface{}) string {
	mJSON, err := json.Marshal(v)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return ""
	}
	msg := string(mJSON)
	return msg
}

// sendMsgToRemote def
func (control Controller) sendMsgToRemote(clientMsg SendRemoteMsg, ch chan string) {
	msg := control.transToStr(clientMsg.msg)
	dataMap := control.jsonToMap(msg)

	// 获取消息体中的traceID，获取不到则使用发送结构体里的traceID字段
	var traceID string
	if tmpID, ok := dataMap["trace_id"]; ok {
		traceID = fmt.Sprintf("%v", tmpID)
	}
	if traceID == "" {
		traceID = clientMsg.traceID
	}

	conn := clientMsg.conn
	err := wsjson.Write(clientMsg.ctx, conn, clientMsg.msg)
	if err != nil {
		var closeErr websocket.CloseError
		if errors.As(err, &closeErr) {
			qflog.Sugare.Warnf("TraceID=%s|conn already closed, client addr=%s", traceID, clientMsg.clientAddr)
		}
		ch <- clientMsg.clientAddr
		qflog.Sugare.Errorf("TraceID=%s|client addr=%s, err:%v", traceID, clientMsg.clientAddr, err)
	}
	qflog.Sugare.Infof("TraceID=%s|send data to server, data=%s, client addr=%s", traceID, msg, clientMsg.clientAddr)
}

func (control Controller) pushSendMsg(msgSt RecvRemoteMsg, sendMsg interface{}) {
	var sendMsgSt SendRemoteMsg
	sendMsgSt.conn = msgSt.conn
	sendMsgSt.clientAddr = msgSt.clientAddr
	sendMsgSt.msg = sendMsg
	sendMsgSt.ctx = msgSt.ctx
	sendMsgSt.traceID = msgSt.traceID
	sendMsgChan <- sendMsgSt
}

// sendDataLoop def
func (control Controller) sendDataLoop(ch chan SendRemoteMsg, closeCh chan string) {
	for msgSt := range ch {
		control.sendMsgToRemote(msgSt, closeCh)
	}
}

// jsonToMap def
func (control Controller) jsonToMap(str string) map[string]interface{} {

	var tempMap map[string]interface{}
	err := json.Unmarshal([]byte(str), &tempMap)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return tempMap
	}

	return tempMap
}

func (control Controller) uploadLogFile(filePath string, dstDirPath string, reportType string) (string, error) {
	server := qfcfg.ServerConfig.Server
	wsse := util.GenWsse(server.Username, server.Password, control.genTraceID())
	url := fmt.Sprintf("http://%s:%d/cava3/upload", server.HttpIP, server.Httpport)

	headers := map[string]string{
		"X-WSSE":            wsse,
		"UPLOAD-DIR-PATH":   dstDirPath,
		"FORCE-UPLOAD":      "TRUE",
		"CAVA3-REPORT-TYPE": reportType,
	}

	dataResp, err := util.HttpPostFile(filePath, url, headers, 20*time.Second)
	if err != nil {
		qflog.Sugare.Errorf("post xls file failed err: %v", err)
		return "", err
	}

	retResp, err := json.Marshal(dataResp)
	if err != nil {
		qflog.Sugare.Errorf("marshal resp err: %v", err)
		return "", err
	}

	var resp JSONResult
	err = json.Unmarshal(retResp, &resp)
	if err != nil {
		qflog.Sugare.Errorf("unmarshal resp err: %v", err)
		return "", err
	}

	if !strings.Contains(resp.Data, "upload succeed") {
		return "", errors.New("upload fail")
	}

	pos := strings.Index(resp.Data, "path:")
	path := resp.Data[pos+len("path:"):]
	return path, nil
}

func (control Controller) convertFile(filePath string, dstDirPath string) (string, error) {
	server := qfcfg.ServerConfig.Server
	wsse := util.GenWsse(server.Username, server.Password, control.genTraceID())
	url := fmt.Sprintf("http://%s:%d/cava3/convert", server.HttpIP, server.Httpport)

	headers := map[string]string{
		"X-WSSE":          wsse,
		"UPLOAD-DIR-PATH": dstDirPath,
		"FORCE-UPLOAD":    "TRUE",
		"OUTPUT-FORMAT":   "htm",
	}

	dataResp, err := util.HttpPostFile(filePath, url, headers, 20*time.Second)
	if err != nil {
		qflog.Sugare.Errorf("post xls file failed err: %v", err)
		return "", err
	}

	retResp, err := json.Marshal(dataResp)
	if err != nil {
		qflog.Sugare.Errorf("marshal resp err: %v", err)
		return "", err
	}

	var resp JSONResult
	err = json.Unmarshal(retResp, &resp)
	if err != nil {
		qflog.Sugare.Errorf("unmarshal resp err: %v", err)
		return "", err
	}

	if !strings.Contains(resp.Data, "convert succeed") {
		return "", errors.New("upload fail")
	}

	pos := strings.Index(resp.Data, "path:")
	path := resp.Data[pos+len("path:"):]
	return path, nil
}

func (control Controller) doTaskWithTimeout(sysType string, cmdStr string, timeout int) (string, string, string, TaskSummaryInfo) {
	var summaryInfo TaskSummaryInfo
	tag := "Excel Report Path: "
	tagLog := "Log:     "
	tagRate := "result pass rate: "
	tagSummary := "result pass summary: "
	xlsData := ""
	logData := ""

	doneChan := make(chan bool, 1)
	errorChan := make(chan error, 1)
	cmd := exec.Command("/bin/bash", "-c", cmdStr)
	pathTag := "/"
	codeType := UTF8
	if sysType == "windows" {
		cmd = exec.Command("cmd", "/C", cmdStr)
		pathTag = "\\"
		codeType = GB18030
	}

	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	go func() {
		//StdoutPipe方法返回一个在命令Start后与命令标准输出关联的管道。Wait方法获知命令结束后会关闭这个管道，一般不需要显式的关闭该管道。
		stdout, err := cmd.StdoutPipe()
		if err != nil {
			qflog.Sugare.Errorf("do task exec cmd err: %v, cmd: %s", err, cmdStr)
			errorChan <- err
			return
		}
		_ = cmd.Start()

		reader := bufio.NewReader(stdout)
		// 实时循环读取输出流中的一行内容
		for {
			bytes, err2 := reader.ReadBytes('\n')
			line := ConvertByte2String(bytes, codeType)
			if err2 != nil || io.EOF == err2 {
				break
			}
			line = strings.TrimRight(line, "\n")
			if sysType == "windows" {
				line = strings.TrimRight(line, "\r")
			}
			qflog.Sugare.Infof(line)

			pos := strings.Index(line, tag)
			if pos != -1 {
				if pos2 := strings.Index(line, ".xls"); pos2 != -1 {
					xlsData = line[pos+len(tag):]
					qflog.Sugare.Infof("xlsData: %s", xlsData)
				}
			}

			pos = strings.Index(line, tagLog)
			if pos != -1 {
				if pos2 := strings.Index(line, ".html"); pos2 != -1 {
					logData = line[pos+len(tagLog):]
					qflog.Sugare.Infof("logData: %s", logData)
				}
			}

			pos = strings.Index(line, tagRate)
			if pos != -1 {
				summaryInfo.PassRate = line[pos+len(tagRate):]
				qflog.Sugare.Infof("pass rate: %s", summaryInfo.PassRate)
			}

			pos = strings.Index(line, tagSummary)
			if pos != -1 {
				summaryInfo.PassSummary = line[pos+len(tagSummary):]
				qflog.Sugare.Infof("pass summary: %s", summaryInfo.PassSummary)
			}
		}
		//阻塞直到该命令执行完成，该命令必须是被Start方法开始执行的
		_ = cmd.Wait()
		doneChan <- true
	}()

	var succeed bool
	select {
	case <-time.After(time.Duration(timeout) * time.Second):
		qflog.Sugare.Infof("%s task timeout", sysType)
		_ = cmd.Process.Kill()
		summaryInfo.PassRate = "0.00%"
		summaryInfo.PassSummary = fmt.Sprintf("%s task timeout", sysType)
		succeed = false
	case <-doneChan:
		qflog.Sugare.Infof("%s task stop", sysType)
		succeed = true
	case err := <-errorChan:
		qflog.Sugare.Errorf("%s exec cmd failed, err: %v", sysType, err)
		summaryInfo.PassRate = "0.00%"
		summaryInfo.PassSummary = fmt.Sprint("%s exec cmd failed, err: %v", sysType, err)
		succeed = false
	}
	if !succeed {
		return "", "", "", summaryInfo
	}

	// xlsData: /usr/local/bin/CAVA3/CAVA_SPACE3/outputs/Log/Creat_CRT_IniFile_Tool_src_20201011194111/Creat_CRT_IniFile_Tool_src_20201011-194124_Test_Report.xls
	xlsData = strings.Trim(xlsData, "\n")
	pos1 := strings.Index(xlsData, fmt.Sprintf("%sLog%s", pathTag, pathTag))
	if pos1 == -1 {
		qflog.Sugare.Error("task output doesn't contains /Log/")
		return "", "", "", summaryInfo
	}

	// tmpPath: Creat_CRT_IniFile_Tool_src_20201011194111/Creat_CRT_IniFile_Tool_src_20201011-194124_Test_Report.xls
	tmpPath := xlsData[pos1+5:]
	pos2 := strings.Index(tmpPath, pathTag)
	dirName := tmpPath[:pos2]
	pos3 := strings.Index(xlsData, dirName)
	dirPath := xlsData[:pos3+len(dirName)]
	pos4 := strings.LastIndex(xlsData, pathTag)
	xlsName := xlsData[pos4+1:] // .xls 文件名
	pos4 = strings.LastIndex(logData, pathTag)
	logName := logData[pos4+1:] // log .html 文件名

	retXls, err := control.uploadLogFile(dirPath+"/"+xlsName, "cava3/outputs/log/"+dirName, "")
	if err != nil {
		qflog.Sugare.Errorf("upload xls file failed err: %v", err)
		return "", "", "", summaryInfo
	}
	qflog.Sugare.Infof("retXls: %s", retXls)

	retLog, err := control.uploadLogFile(dirPath+"/"+logName, "cava3/outputs/log/"+dirName, "log")
	if err != nil {
		qflog.Sugare.Errorf("upload log file failed err: %v", err)
		return "", "", "", summaryInfo
	}

	// report html 文件名可由 log html 替换而来
	reportName := strings.Replace(logName, "log.html", "report.html", -1)
	_, _ = control.uploadLogFile(dirPath+"/"+reportName, "cava3/outputs/log/"+dirName, "report")
	qflog.Sugare.Infof("retLog: %s", retLog)

	retHtml, err := control.convertFile(dirPath+"/"+xlsName, "cava3/outputs/log/"+dirName)
	if err != nil {
		qflog.Sugare.Errorf("convert xls file to html failed err: %v", err)
	}
	qflog.Sugare.Infof("retHtml: %s", retHtml)
	return retXls, retLog, retHtml, summaryInfo
}

// ConvertByte2String def
func ConvertByte2String(byte []byte, charset Charset) string {
	var str string
	switch charset {
	case GB18030:
		var decodeBytes, _ = simplifiedchinese.GB18030.NewDecoder().Bytes(byte)
		str = string(decodeBytes)
	case UTF8:
		fallthrough
	default:
		str = string(byte)
	}
	return str
}

func (control Controller) reportToSmu(msgSt RecvRemoteMsg, xlsPth string, logPth string, htmlPath string, summaryInfo TaskSummaryInfo, reqSt TaskReqMsg) {
	// 待发送的消息写入对应channel
	var reqMsgSt TaskReportReqMsg
	reqMsgSt.MsgCmd = cavaTaskReportNotifyReqMsg
	reqMsgSt.TaskID = reqSt.TaskID
	reqMsgSt.TaskReport = xlsPth
	reqMsgSt.TaskLog = logPth
	reqMsgSt.TaskHtml = htmlPath
	reqMsgSt.PassRate = summaryInfo.PassRate
	reqMsgSt.PassSummary = summaryInfo.PassSummary
	reqMsgSt.TaskName = reqSt.TaskName
	reqMsgSt.NetID = "cava_teu"
	reqMsgSt.TeuID = qfcfg.ServerConfig.Client.TeuID
	reqMsgSt.TraceID = reqSt.TraceID

	control.pushSendMsg(msgSt, reqMsgSt)
}

// path file existt
func pathFileExists(path string) (bool, error) {
	_, err := os.Stat(path)
	if err == nil {
		// qflog.Sugare.Infof("file:%s exist.", path)
		return true, nil
	}
	if os.IsNotExist(err) {
		// qflog.Sugare.Errorf("file:%s not exist.", path)
		return false, err
	}
	return false, err
}

func (control Controller) checkTaskReqParam(req *TaskReqMsg) error {
	if exist, _ := pathFileExists(req.TaskDir); !exist {
		return errors.New("task dir not exist")
	}

	// task 执行默认超时时间
	if req.Timeout == 0 {
		req.Timeout = qfcfg.ServerConfig.Client.Timeout
	}

	if req.Timeout <= 0 {
		return errors.New("task timeout must more than 0")
	}

	if req.TestData == "" {
		return nil
	}

	fileName := util.BaseName(req.TestData)
	if !strings.HasSuffix(fileName, ".xls") {
		return errors.New("file format is not xls")
	}

	_, err := util.DownloadFile(req.TestData, req.TaskDir, "DeviceList.xls")
	if err != nil {
		return errors.Wrap(err, "download file er")
	}

	return nil
}

// doTaskReqMsg def
func (control Controller) doTaskReqMsg(msgSt RecvRemoteMsg) {
	msg := control.transToStr(msgSt.msg)
	var reqMsgSt TaskReqMsg
	_ = json.Unmarshal([]byte(msg), &reqMsgSt)

	// 待发送的消息写入对应channel
	var respMsgSt TaskRespMsg
	respMsgSt.MsgCmd = cavaTaskRespMsg
	respMsgSt.TaskID = reqMsgSt.TaskID
	respMsgSt.TaskName = reqMsgSt.TaskName
	respMsgSt.NetID = "cava_teu"
	respMsgSt.Code = 200
	respMsgSt.RetMsg = "ok"
	respMsgSt.TraceID = reqMsgSt.TraceID

	cmd := ""
	xlsPath := ""
	logPath := ""
	reportPath := ""
	var summaryInfo TaskSummaryInfo
	if err := control.checkTaskReqParam(&reqMsgSt); err != nil {
		respMsgSt.Code = 400
		respMsgSt.RetMsg = err.Error()
		control.pushSendMsg(msgSt, respMsgSt)
	} else {
		if strings.ToLower(reqMsgSt.SysType) == "linux" {
			cmd = fmt.Sprintf("source /etc/profile&&cd  %s&&sh run.sh", reqMsgSt.TaskDir)
		} else if strings.ToLower(reqMsgSt.SysType) == "windows" {
			cmd = fmt.Sprintf("%s\\run.bat", reqMsgSt.TaskDir)
		}
		qflog.Sugare.Infof("TraceID=%s|cmd=%s, TestData=%s, SysType=%s, timeout=%d,", reqMsgSt.TraceID, cmd, reqMsgSt.TestData, reqMsgSt.SysType, reqMsgSt.Timeout)

		control.pushSendMsg(msgSt, respMsgSt)
		xlsPath, logPath, reportPath, summaryInfo = control.doTaskWithTimeout(strings.ToLower(reqMsgSt.SysType), cmd, reqMsgSt.Timeout)
	}

	control.reportToSmu(msgSt, xlsPath, logPath, reportPath, summaryInfo, reqMsgSt)
	qflog.Sugare.Infof("TraceID=%s|do task goroutine exit, TaskID=%s", reqMsgSt.TraceID, reqMsgSt.TaskID)
}

// AliveCount export var
var AliveCount int

// doKeepAlive def
func (control Controller) doKeepAlive(msgSt RecvRemoteMsg, closeCh chan string) {
	msg := control.transToStr(msgSt.msg)
	var registerRespMsgSt RegisterRespMsg
	_ = json.Unmarshal([]byte(msg), &registerRespMsgSt)

	if registerRespMsgSt.Code != 200 {
		qflog.Sugare.Errorf("traceID=%s|teu register to smu failed, code:%d client addr:%s", msgSt.traceID, registerRespMsgSt.Code, msgSt.clientAddr)
		return
	}

	exitFlag := false
	for {
		if exitFlag {
			qflog.Sugare.Warnf("TraceID=%s|conn closed, exit keep alive, client addr=%s", msgSt.traceID, msgSt.clientAddr)
			break
		}
		// if AliveCount < 6 {
		// 	// 待发送的消息写入对应channel
		// 	var reqMsgSt KeepAliveReqMsg
		// 	reqMsgSt.MsgCmd = cavaKeepAliveReqMsg
		// 	reqMsgSt.NetID = "cava_teu"
		// 	control.pushSendMsg(msgSt, reqMsgSt)
		// } else if AliveCount == 6 {
		// 	qflog.Sugare.Warnf("traceID=%s|teu heartbeat to smu stop, client addr:%s", msgSt.traceID, msgSt.clientAddr)
		// }

		// 待发送的消息写入对应channel
		var reqMsgSt KeepAliveReqMsg
		reqMsgSt.MsgCmd = cavaKeepAliveReqMsg
		reqMsgSt.NetID = "cava_teu"
		reqMsgSt.TraceID = control.checkTraceID()

		// 超时机制
		timeout := make(chan bool, 1)
		go func() {
			time.Sleep(30 * time.Second)
			timeout <- true
		}()

		select {
		case ch := <-closeCh:
			exitFlag = msgSt.clientAddr == ch
		case <-timeout:
			control.pushSendMsg(msgSt, reqMsgSt)
			AliveCount++
		}
	}
}

// registerToSmu def
func (control Controller) registerToSmu(ctx context.Context, conn *websocket.Conn) {

	traceID := control.checkTraceID()

	// 构造1个假的接收消息结构体，方便向smu主动发送消息
	var msgSt RecvRemoteMsg
	msgSt.conn = conn
	msgSt.clientAddr = clientAddr
	msgSt.msg = nil
	msgSt.ctx = ctx
	msgSt.traceID = traceID

	// 待发送的消息写入对应channel
	var reqMsgSt RegisterReqMsg
	reqMsgSt.MsgCmd = cavaRegisterReqMsg
	reqMsgSt.NetID = "cava_teu"
	reqMsgSt.TeuID = qfcfg.ServerConfig.Client.TeuID
	reqMsgSt.SysType = qfcfg.ServerConfig.Client.SysType
	reqMsgSt.TraceID = control.checkTraceID()

	control.pushSendMsg(msgSt, reqMsgSt)
}

// processData def
func (control Controller) processData(recvMsg RecvRemoteMsg, closeCh chan string) {
	msg := control.transToStr(recvMsg.msg)
	dataMap := control.jsonToMap(msg)

	// 获取消息体中的traceID，获取不到则临时生成1个
	var traceID string
	if tmpID, ok := dataMap["trace_id"]; ok {
		traceID = fmt.Sprintf("%v", tmpID)
	}
	if traceID == "" {
		traceID = control.checkTraceID()
	}

	if cmd, ok1 := dataMap["msg_cmd"]; ok1 {
		// interface{} 间接转int
		cmdStr := fmt.Sprintf("%v", cmd)
		cmdVal, _ := strconv.Atoi(cmdStr)
		qflog.Sugare.Infof("TraceID=%s|processData, cmd=%d, type=%s, recv data from server, data=%s, client addr=%s", traceID,
			cmdVal, msgTypeMap[cmdVal], msg, recvMsg.clientAddr)

		switch cmdVal {
		case cavaTaskReqMsg:
			go control.doTaskReqMsg(recvMsg)
		case cavaRegisterRespMsg:
			go control.doKeepAlive(recvMsg, closeCh)
		}
	}
}

// recvDataLoop def
func (control Controller) recvDataLoop(ch chan RecvRemoteMsg, closeCh chan string) {
	for msgSt := range ch {
		control.processData(msgSt, closeCh)
	}
}

var clientAddr string

// readData def
func (control Controller) readDataLoop(ctx context.Context, conn *websocket.Conn, ch chan string) {
	for {
		var v interface{}
		err := wsjson.Read(ctx, conn, &v)
		if err != nil {
			var closeErr websocket.CloseError
			if errors.As(err, &closeErr) {
				qflog.Sugare.Warnf("conn already closed, client addr=%s", clientAddr)
			}
			ch <- clientAddr
			qflog.Sugare.Errorf("err:%v", err)
			break
		}

		msg := control.transToStr(v)
		dataMap := control.jsonToMap(msg)
		// 获取消息体中的traceID，获取不到则临时生成1个
		var traceID string
		if tmpID, ok := dataMap["trace_id"]; ok {
			traceID = fmt.Sprintf("%v", tmpID)
		}
		if traceID == "" {
			traceID = control.genTraceID()
		}

		// 解析客户端地址，赋值给全局变量
		if clientAddr == "" {
			ok := false
			if _, ok = dataMap["clientAddr"]; ok {
				clientAddr = fmt.Sprintf("%v", dataMap["clientAddr"])
				qflog.Sugare.Infof("TraceID=%s|client addr:%s", traceID, clientAddr)
				// 向smu注册
				control.registerToSmu(ctx, conn)
			}
		}

		// 接收到的消息写入对应channel
		var msgSt RecvRemoteMsg
		msgSt.conn = conn
		msgSt.clientAddr = clientAddr
		msgSt.msg = v
		msgSt.ctx = ctx
		msgSt.traceID = traceID
		recvMsgChan <- msgSt
	}
}

// Start def
func (control Controller) Start() {
	traceID := control.checkTraceID()
	clientAddr = ""

	ctx := context.Background()
	// ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	// defer cancel()

	url := fmt.Sprintf("ws://%s:%d/ws", qfcfg.ServerConfig.Server.IP, qfcfg.ServerConfig.Server.Port)
	conn, _, err := websocket.Dial(ctx, url, nil)
	if err != nil {
		qflog.Sugare.Errorf("err:%v", err)
		return
	}

	defer conn.Close(websocket.StatusInternalError, "内部错误")
	qflog.Sugare.Infof("TarceID=%s|client connect succeed, server addr=%s:%d", traceID, qfcfg.ServerConfig.Server.IP, qfcfg.ServerConfig.Server.Port)

	// 开启读数据协程
	go control.readDataLoop(ctx, conn, closeCh)

	// 如果获取到的 close 消息是当前链路，则关闭
	var flag bool
	for ch := range closeCh {
		flag = ch == clientAddr
		if flag {
			break
		}
	}

	qflog.Sugare.Warnf("TarceID=%s|exit go routinue: %v, client addr=%s", traceID, flag, clientAddr)
}
